package com.dxf.bigdata.D05_spark_again.存储

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * persist  放硬盘或者内存,有级别
 * 默认放内存  就是cache()
 * StorageLevel
 *
 * 持久化作用,保证数据不丢失,可以重用RDD的结果,也适合不稳定的情况下
 *
 */
object persist {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("app")

    sparkConf.set("spark.port.maxRetries","100")

    val sc = new SparkContext(sparkConf)

    //word count
    val rdd: RDD[String] = sc.makeRDD(List("hello spark", "hello word"))

    val splitRDD: RDD[String] = rdd.flatMap(_.split(" "))

    val mapRDD: RDD[(String, Int)] = splitRDD.map(x => {
      println("map ---")
      (x, 1)
    })
    //持久化
    val persistRDD: mapRDD.type = mapRDD.persist()
    val value: RDD[(String, Int)] = persistRDD.reduceByKey(_ + _)

    value.collect().foreach(println)

    println("other ===================")

    // mapRDD 再次使用,RDD是不存储数据的,所以重新计算一遍; mapRDD通过血缘关系重复计算
    val value1: RDD[(String, Iterable[Int])] = persistRDD.groupByKey()

    value1.collect().foreach(println)
    sc.stop()

  }

}
